package com.atguigu.chapter06;

import com.atguigu.Bean.MarketingUserBehavior;
import javassist.CtClass;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;
import sun.font.TrueTypeFont;

import java.util.HashMap;
import java.util.Random;

/**
 * @ClassName: Flink02_AppAnalysis_By_Chanel
 * @Description:
 * @Author: kele
 * @Date: 2021/4/6 9:29
 *
 *  APP市场推广统计 - 分渠道
 * 不同渠道，行为统计
 **/
public class Flink02_AppAnalysis_By_Chanel {

    public static void main(String[] args) {



        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        //获取数据
        DataStreamSource<MarketingUserBehavior> ds = env.addSource(new SourceFunction<MarketingUserBehavior>() {

            Random random = new Random();
            Boolean flag = true;

            String[] behaviors = {"download","uninstall","install","update"};
            String[] channel = {"hauwei", "xiaomi", "iphone", "oppo"};

            @Override
            public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {

                while (flag) {
                    ctx.collect(new MarketingUserBehavior(
                            (long) random.nextInt(200),
                            behaviors[random.nextInt(behaviors.length)],
                            channel[random.nextInt(channel.length)],
                            System.currentTimeMillis()
                    ));

                    Thread.sleep(2000);
                }
            }

            @Override
            public void cancel() {
                flag = false;

            }
        });

        /**
         * 1、先通过keyby将相同的key进入到同一个并行度下
         * 2、通过process返回元组（word，count）的形式
         *      --不能使用reduce的形式，返回的数据类型改变了
         */

        ds.keyBy( mub -> mub.getBehavior()+"_"+mub.getChannel())
                //reduce的返回值类型和输出的类型一致，所以不能使用
        .process(new KeyedProcessFunction<String, MarketingUserBehavior, Tuple2<String,Long>>() {

            //不能使用sum，sum的个数和并行度相关，

            //使用map时，跟并行度相关，但同一个key在同一个并行度下，可以得到正确的结果，
            HashMap<String,Long> market = new HashMap<>();

            @Override
            public void processElement(MarketingUserBehavior value, Context ctx, Collector<Tuple2<String, Long>> out) throws Exception {

                if (!market.containsKey(value.getBehavior() + "_" + value.getChannel())) {

                    market.put(value.getBehavior() + "_" + value.getChannel(),0l);

                }

                market.put(value.getBehavior() + "_" + value.getChannel(),
                        market.get(value.getBehavior() + "_" + value.getChannel()) + 1);

                out.collect(Tuple2.of(ctx.getCurrentKey(), market.get(ctx.getCurrentKey())));

            }
        }).print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

}
